// 一个通道应该支持多个发送者和接收者。
// 在实际使用中，我们需要使用不同的库来满足诸如：多发送者 -> 单接收者，多发送者 -> 多接收者等场景形式
// 对于通道而言，消息的发送顺序和接收顺序是一致的，满足FIFO原则(先进先出)
// 多(单)个发送者，但是只支持唯一的接收者
#[cfg(test)]
mod mpsc_channel {
    use std::sync::mpsc;
    use std::sync::mpsc::{Receiver, Sender};
    use std::thread;
    use std::time::Duration;

    #[test]
    fn main() {
        // 创建一个消息通道, 返回一个元组：(发送者，接收者)
        let (tx, rx) = mpsc::channel(); //tx,rx对应发送者和接收者，它们的类型由编译器自动推导(发送的数据类型)

        // 创建线程，并发送消息，要使用move将tx的所有权转移到子线程的闭包中
        thread::spawn(move || {
            // 发送一个数字1, send方法返回Result<T,E>，通过unwrap进行快速错误处理
            tx.send(1).unwrap();
            // 下面代码将报错，因为编译器自动推导出通道传递的值是i32类型，那么Option<i32>类型将产生不匹配错误
            // tx.send(Some(1)).unwrap()
        });
        // 接收消息的操作rx.recv()会阻塞当前线程，直到读取到值，或者通道被关闭
        // 在主线程中接收子线程发送的消息并输出
        // println!("receive {}", rx.recv().unwrap());

        // 使用try_recv尝试接收一次消息，该方法并不会阻塞线程，当通道中没有消息时，它会立刻返回一个错误
        println!("receive {:?}", rx.try_recv()); //Err(Empty)
        thread::sleep(1 * Duration::from_millis(200));
        println!("receive {:?}", rx.try_recv()); //OK(1)
        thread::sleep(1 * Duration::from_millis(200));
        println!("receive {:?}", rx.try_recv()); //Err(Disconnected) 发送者已经被关闭。
    }

    /*
    传输具有所有权的数据
        若值的类型实现了Copy特征，则直接复制一份该值，然后传输过去，例如之前的i32类型
        若值没有实现Copy，则它的所有权会被转移给接收端，在发送端继续使用该值将报错
    */
    #[test]
    fn channel_mv() {
        let (tx, rx) = mpsc::channel();

        thread::spawn(move || {
            //String底层的字符串是存储在堆上，并没有实现Copy特征
            let s = String::from("我，飞走咯!");

            tx.send(s).unwrap(); //当它被发送后，会将所有权从发送端的s转移给接收端的received
            // println!("val is {}", s); //err 无法再使用s
        });

        let received = rx.recv().unwrap();
        println!("Got: {}", received);
    }

    #[test]
    fn chan_for_receive() {
        let (tx, rx) = mpsc::channel();
        thread::spawn(move || {
            //主线程和子线程是并发运行的，子线程在不停的发送消息
            let vals = vec![
                String::from("hi"),
                String::from("from"),
                String::from("the"),
                String::from("thread"),
            ];
            for val in vals {
                tx.send(val).unwrap();
                thread::sleep(Duration::from_millis(200));
            }
        });
        //主线程使用for循环 阻塞 的从rx迭代器中接收消息
        //当子线程运行完成时，发送者tx会随之被drop，此时for循环将被终止，最终main线程成功结束。
        for received in rx {
            println!("Got: {}", received);
        }
    }
    /*

    需要所有的发送者都被drop掉后，接收者rx才会收到错误，进而跳出for循环，最终结束主线程
    这里虽然用了clone但是并不会影响性能，因为它并不在热点代码路径中，仅仅会被执行一次
    由于两个子线程谁先创建完成是未知的，因此哪条消息先发送也是未知的，最终主线程的输出顺序也不确定

    */
    #[test]
    fn chan_more_send() {
        let (tx, rx) = mpsc::channel();
        let tx1 = tx.clone(); //一个对发送者的克隆let tx1 = tx.clone();
        thread::spawn(move || {
            tx.send(String::from("hi from raw tx")).unwrap();
        });

        thread::spawn(move || {
            tx1.send(String::from("hi from cloned tx")).unwrap();
        });

        for received in rx {
            println!("Got: {}", received);
        }
    }

    // 异步通道：无论接收者是否正在接收消息，消息发送者在发送消息时都不会阻塞:
    #[test]
    fn chan_asyn() {
        // 异步
        let (tx, rx) = mpsc::channel();

        let handle = thread::spawn(move || {
            // 发送之前和发送之后是连续输出的，没有受到接收端主线程的任何影响，
            // 因此通过mpsc::channel创建的通道是异步通道

            println!("发送之前");
            tx.send(1).unwrap();
            println!("发送之后");
        });

        println!("睡眠之前");
        thread::sleep(Duration::from_millis(1 * 200));
        println!("睡眠之后");

        // 主线程因为睡眠阻塞了 3 秒，因此并没有进行消息接收，而子线程却在此期间轻松完成了消息的发送。
        // 等主线程睡眠结束后，才姗姗来迟的从通道中接收了子线程老早之前发送的消息。
        println!("receive {}", rx.recv().unwrap());
        handle.join().unwrap();
    }

    // 同步通道发送消息是阻塞的，只有在消息被接收后才解除阻塞
    #[test]
    fn chan_sync() {
        // sync_channel
        let (tx, rx) = mpsc::sync_channel(0); // go: make (chan int,0)

        let handle = thread::spawn(move || {
            println!("发送之前");
            tx.send(1).unwrap(); // 接收端未接收 导致阻塞
            println!("发送之后");
        });

        println!("睡眠之前");
        thread::sleep(Duration::from_millis(3 * 100));
        println!("睡眠之后");

        println!("receive {}", rx.recv().unwrap());
        handle.join().unwrap();
    }

    #[test]
    fn chan_sync_buf() {
        // sync_channel
        let (tx, rx) = mpsc::sync_channel(2); // go: make (chan int,0)

        let handle = thread::spawn(move || {
            for x in 1..=3 {
                println!("发送之前 {}", x);
                tx.send(x).unwrap(); // 接收端未接收 导致阻塞
                println!("发送之后 {}", x);
            }
        });

        println!("睡眠之前");
        thread::sleep(Duration::from_millis(3 * 100));
        println!("睡眠之后");

        // 所有发送者被drop或者所有接收者被drop后，通道会自动关闭
        for received in rx {
            println!("receive {}", received);
        }
        handle.join().unwrap();
    }

    // Rust 会按照枚举中占用内存最大的那个成员进行内存对齐，
    // 这意味着就算你传输的是枚举中占用内存最小的成员，它占用的内存依然和最大的成员相同, 因此会造成内存上的浪费
    enum Fruit {
        Apple(u8),
        Orange(String),
    }

    #[test]
    fn chan_msg_type() {
        let (tx, rx): (Sender<Fruit>, Receiver<Fruit>) = mpsc::channel();

        tx.send(Fruit::Orange("sweet".to_string())).unwrap();
        tx.send(Fruit::Apple(2)).unwrap();

        for _ in 0..2 {
            match rx.recv().unwrap() {
                Fruit::Apple(count) => println!("received {} ", count),
                Fruit::Orange(flavor) => println!("received {} ", flavor),
            }
        }
    }

    #[test]
    fn mpsc_err() {
        use std::thread;
        let (send, recv) = mpsc::channel();
        let num_threads = 3;
        for i in 0..num_threads {
            // 子线程拿走的是复制后的send的所有权，这些拷贝会在子线程结束后被drop
            // 但是send本身却直到main函数的结束才会被drop
            let thread_send = send.clone();
            thread::spawn(move || {
                thread_send.send(i).unwrap();
                println!("thread {:?} finished", i);
            });
        }

        //通道关闭的两个条件：发送者全部drop或接收者被drop，
        // 要结束for循环显然是要求发送者全部drop，但由于send自身没有被drop，会导致该循环永远无法结束，最终主线程会一直阻塞

        drop(send); // 在这里drop send...通道关闭
        for x in recv {
            println!("Got: {}", x);
        }
        println!("finished iterating");
    }
}

#[cfg(test)]
mod mpsc_example {
    use std::thread;
    use std::time::Duration;

    #[test]
    fn main() {
        // 创建一个线程A
        let new_thread = thread::spawn(move || {
            println!("new_thread start!");
            // 再创建一个线程B
            let handle = thread::spawn(move || {
                println!("child start!");
                thread::sleep(Duration::from_millis(200));
                println!("child finish!");
            });

            handle.join().unwrap();

            println!("new_thread finish!");
        });

        // 等待新创建的线程执行完成
        new_thread.join().unwrap();
        println!("join thread is start!!");

        // 睡眠一段时间，看子线程创建的子线程是否还在运行
        thread::sleep(Duration::from_millis(100));
        println!("main end!")
    }
}
